﻿using EventBus;
using EventBus.Abstractions;
using Newtonsoft.Json;
using Polly;
using RabbitMQ.Client;
using RabbitMQ.Client.Events;
using RabbitMQ.Client.Exceptions;
using System;
using System.Collections.Generic;
using System.Net.Sockets;
using System.Text;

namespace EventBusRabbitMQ
{
    /// <summary>
    /// Rabbit mq event bus.
    /// </summary>
    public class EventBusRabbitMQ
        : AbstractEventBus
    {
        private readonly IConnectionFactory _connectionFactory;
        private IConnection _connection;
        private IModel _channel;
        private readonly string _exchangeType;
        private readonly string _exchangeName;
        private readonly string _queueName;
        private readonly bool _autoAck;
        public EventBusRabbitMQ(IConnectionFactory connectionFactory, string exchangeName, string exchangeType, string queneName, bool autoAck, IEventHandlerExecutionContext eventExecutionContext) :
            base(eventExecutionContext)
        {
            _connectionFactory = connectionFactory;
            _connection = connectionFactory.CreateConnection();
            _exchangeName = exchangeName;
            _exchangeType = exchangeType;
            _queueName = queneName;
            _autoAck = autoAck;

            CreateConsumerChannel();
        }


        public override void Publish<TEvent>(TEvent @event)
        {
            CheckAndConnect();

            using (var channel = _connection.CreateModel())
            {
                channel.ExchangeDeclare(_exchangeName, _exchangeType, true);

                var itemBody = Encoding.UTF8.GetBytes(JsonConvert.SerializeObject(@event));
                channel.BasicPublish(_exchangeName, typeof(TEvent).FullName, body: itemBody);
            }
        }


        public override void Subscribe<TEvent, TEventHandler>()
        {
            if (!_eventHandlerExecutionContext.IsRegisteEventHandler<TEvent, TEventHandler>())
            {
                _eventHandlerExecutionContext.RegisteEventHandler<TEvent, TEventHandler>();
                _channel.QueueBind(_queueName, _exchangeName, typeof(TEvent).FullName, null);
            }
        }


        /// <summary>
        /// Check and connect.
        /// </summary>
        private void CheckAndConnect()
        {
            var policy = Policy.Handle<SocketException>()
                .Or<BrokerUnreachableException>()
                .WaitAndRetry(3, c => TimeSpan.FromSeconds(3));

            policy.Execute(() =>
            {
                if (_connection != null && _connection.IsOpen)
                {
                    _connection = _connectionFactory.CreateConnection();
                }
            });
        }


        /// <summary>
        ///Create consumer channel. 
        /// </summary>
        private void CreateConsumerChannel()
        {
            _channel = _connection.CreateModel();
            _channel.ExchangeDeclare(_exchangeName, _exchangeType, true);
            _channel.QueueDeclare(_queueName, true, false, false, null);

            var consumer = new EventingBasicConsumer(_channel);
            consumer.Received += async (model, ea) =>
            {
                var itemBody = ea.Body;
                var message = (IEvent)JsonConvert.DeserializeObject(Encoding.UTF8.GetString(itemBody));
                await _eventHandlerExecutionContext.HandleAsync(message);

                _channel.BasicAck(ea.DeliveryTag,false);
            };
        }
    }
}
